package com.steakliu.sun.proxy;

import com.steakliu.sun.cache.result.CacheInfo;
import com.steakliu.sun.cache.result.CacheHelper;
import com.steakliu.sun.cache.result.ResultCacheManager;
import com.steakliu.sun.client.api.ClientConfig;
import com.steakliu.sun.client.netty.NettyClient;
import com.steakliu.sun.common.thread.ClientConcurrentThreadPool;
import com.steakliu.sun.constant.RateLimiterFailStrategyConstant;
import com.steakliu.sun.invoker.api.Invoker;
import com.steakliu.sun.protocol.*;
import com.steakliu.sun.protocol.context.RpcContext;
import com.steakliu.sun.protocol.future.SunFuture;
import com.steakliu.sun.protocol.generate.MsgHeaderGenerator;
import com.steakliu.sun.ratelimiter.api.SunRateLimiter;
import com.steakliu.sun.registry.api.RegistryConfig;
import com.steakliu.sun.registry.api.RegistryService;
import com.steakliu.sun.spi.loader.ExtensionLoader;
import lombok.Data;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.util.concurrent.TimeUnit;

/**
 * 功能说明： 代理对象
 * <p>
 * Original @Author: steakliu-刘牌, 2023-01-08  20:25
 * <p>
 * Copyright (C)2020-2022  steakliu All rights reserved.
 */
@Data
public class ObjectProxy<T> implements InvocationHandler {
    private final Logger LOGGER = LoggerFactory.getLogger(ObjectProxy.class);

    private Class<T> clazz;

    private ClientConfig config;

    private NettyClient nettyClient;

    private ClientConcurrentThreadPool clientConcurrentThreadPool;

    private Invoker invoker;

    private SunRateLimiter rateLimiter;

    private ResultCacheManager resultCacheManager;

    public ObjectProxy(Class<T> clazz, ClientConfig config) throws Exception {
        this.clazz = clazz;
        this.config = config;
        this.clientConcurrentThreadPool = ClientConcurrentThreadPool.getInstance(config.getConcurrentCorePoolSize(), config.getConcurrentMaximumPoolSize());

        RegistryService registryService = ExtensionLoader.getExtension(RegistryService.class, config.getRegistryType());
        registryService.init(new RegistryConfig(config.getRegistryAddress()));

        if (config.isUseRateLimit()) {
            this.rateLimiter = ExtensionLoader.getExtension(SunRateLimiter.class, config.getRateLimiterType());
            this.rateLimiter.init(config.getMilliseconds(), config.getPermits());
        }
        this.nettyClient = NettyClient.getInstance(config, registryService, clientConcurrentThreadPool);
        this.invoker = ExtensionLoader.getExtension(Invoker.class, config.getReflectType());
        this.resultCacheManager = ResultCacheManager.getInstance();
    }

    @Override
    public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
        if (method.getDeclaringClass() == Object.class) {
            String name = method.getName();
            switch (name) {
                case "equals":
                    return proxy == args[0];
                case "hashCode":
                    return System.identityHashCode(proxy);
                case "toString":
                    return proxy.getClass().getName() + "@" + Integer.toHexString(System.identityHashCode(proxy)) + ", with InvocationHandler " + this;
                default:
                    throw new IllegalArgumentException(String.valueOf(method));
            }
        }
        SunProtocol<SunRequest> protocol = new SunProtocol<>();
        SunRequest request = new SunRequest();
        request.setClassName(clazz.getName());
        request.setMethodName(method.getName());
        request.setParameterTypes(method.getParameterTypes());
        request.setParameters(args);
        request.setVersion(config.getVersion());
        request.setGroup(config.getGroup());
        request.setToken(config.getToken());
        protocol.setMsgHeader(MsgHeaderGenerator.generateMsgHeader(config.getSerializer(), MsgType.REQUEST));
        protocol.setBody(request);
        Object result;
        //使用缓存
        if (config.isCache()) {
            result = this.getResultFromCache(request, protocol);
        } else {
            result = this.sendRequest(protocol);
        }
        return result;
    }

    /**
     * 从缓存中获取结果
     *
     * @param request
     * @param protocol
     * @return
     */
    private Object getResultFromCache(SunRequest request, SunProtocol<SunRequest> protocol) {
        String cacheKey = CacheHelper.buildCacheKey(request);
        CacheInfo cacheInfo = resultCacheManager.get(cacheKey);
        Object result;
        if (null == cacheInfo) {
            result = this.sendRequest(protocol);
            if (null != result) {
                CacheInfo cache = new CacheInfo(CacheHelper.getExpiredTimestamp(config.getCacheTime()), result);
                resultCacheManager.put(cacheKey, cache);
            }
        } else {
            result = cacheInfo.getData();
        }
        return result;
    }

    private Object sendRequest(SunProtocol<SunRequest> protocol) {
        try {
            SunFuture future = this.sendRequestOrRateLimiter(protocol);
            return null == future ? null : config.getTimeout() > 0 ? future.get(config.getTimeout(), TimeUnit.MILLISECONDS) : future.get();
        } catch (Throwable e) {
            e.printStackTrace();
            //服务容错处理
            Class<?> fallbackClass = config.getFallbackClass();
            if (void.class != fallbackClass) {
                try {
                    SunRequest request = protocol.getBody();
                    Object serviceBean = fallbackClass.newInstance();
                    return invoker.invoke(serviceBean, fallbackClass, request.getMethodName(), request.getParameterTypes(), request.getParameters());
                } catch (Throwable throwable) {
                    throwable.printStackTrace();
                }
            } else {
                throw new RuntimeException(e);
            }
        }
        return null;
    }

    private SunFuture sendRequestOrRateLimiter(SunProtocol<SunRequest> protocol) throws Exception {
        if (config.isUseRateLimit()) {
            try {
                if (rateLimiter.tryAcquire()) {
                    return this.singleSendRequest(protocol);
                } else {
                    if (config.getRateLimiterFailStrategy().equals(RateLimiterFailStrategyConstant.FALL_BACK)) {
                        //降级处理
                        return this.rateLimiterFailFallback(protocol);
                    } else if (config.getRateLimiterFailStrategy().equals(RateLimiterFailStrategyConstant.RETRY)) {
                        //重试
                        return this.singleSendRequest(protocol);
                    }
                }
            } finally {
                this.rateLimiter.release();
            }
        } else {
            return this.singleSendRequest(protocol);
        }
        return null;
    }

    private SunFuture singleSendRequest(SunProtocol<SunRequest> protocol) throws Exception {
        return nettyClient.send(protocol);
    }

    /**
     * 流量达到限流最大值，触发降级
     *
     * @param protocol
     */
    private SunFuture rateLimiterFailFallback(SunProtocol<SunRequest> protocol) {
        SunProtocol<SunResponse> responseSunProtocol = new SunProtocol<>();
        MsgHeader header = protocol.getMsgHeader();
        header.setType((byte) MsgType.RESPONSE.getType());
        header.setStatus((byte) MsgStatus.FAIL.getStatus());
        responseSunProtocol.setMsgHeader(header);
        SunResponse response = new SunResponse();
        response.setErrorMsg("client trigger service fallback");
        responseSunProtocol.setBody(response);
        SunFuture future = SunFuture.getInstance();
        future.done(responseSunProtocol);
        return future;
    }
}
